package org.citywithincity.jobqueue;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.LinkedBlockingQueue;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import com.citywithincity.db.Db;
import com.damai.plugins.IPlugin;

/**
 * 消费者、生产者
 * 
 * @author Randy
 *
 * @param <T>
 */
public class PcQueue implements IPlugin, JobQueueManager, JobHandlerManager {

	QueueManager manager;
	/**
	 * 任务队列,按照优先级处理任务
	 */
	LinkedBlockingQueue<Object> jobQueue;

	private int threadCount;
	/**
	 * 任务处理器
	 */
	private Map<Class<?>, JobHandler<?>> handlers;
	/**
	 * 线程池
	 */
	protected PcQueueThread[] queueThreads;
	private QueueListener listener;
	private Log logger = LogFactory.getLog(PcQueue.class);
	/**
	 * 是否正在运行
	 */
	private volatile boolean isRunning;

	public PcQueue() {
		manager = QueueManager.getInstance();
		jobQueue = new LinkedBlockingQueue<Object>();
		handlers = new ConcurrentHashMap<Class<?>, JobHandler<?>>();
	}

	public void setThreadCount(int threadCount) {
		this.threadCount = threadCount;
	}

	@Override
	public void start() {
		if (isRunning)
			return;
		queueThreads = new PcQueueThread[threadCount];
		for (int i = 0; i < threadCount; i++) {
			queueThreads[i] = new PcQueueThread(jobQueue, manager);
			queueThreads[i].setHandler(this);
			queueThreads[i].setName(String.format("Pool[%02d]", i));
			queueThreads[i].start();
		}
		isRunning = true;
	}

	@Override
	public void stop() {
		if (!isRunning)
			return;
		for (int i = 0; i < queueThreads.length; i++) {
			queueThreads[i].stop();
			queueThreads[i] = null;
		}
	}

	@Override
	public void add(Object job) {
		jobQueue.add(job);
	}

	public QueueListener getListener() {
		return listener;
	}

	public void setListener(QueueListener listener) {
		this.listener = listener;
	}

	private AsyncJobHandler AsyncJobHandler = new AsyncJobHandler();

	@SuppressWarnings({ "unchecked", "rawtypes" })
	@Override
	public void onExecuteJob(Object job) {
		JobHandler handler;
		
		if (job instanceof AsyncJob) {
			handler = AsyncJobHandler;
		} else {
			handler = handlers.get(job.getClass());
		}
		
		if (handler == null) {
			logger.warn("Job " + job.getClass().getName() + " has no handler.");
			return;
		}
		
		long time = System.currentTimeMillis();
		logger.info("正在执行异步任务" + job);
		try {
			handler.onAsyncJob(job);
		} catch (Throwable e) {
			// 将job保存到文件上面,并排查原因
			logger.error("onExecuteJobError " + job.toString(), e);
		} finally {
			logger.info(String.format("异步任务执行完成%s,耗时%d", 
					job.toString(), 
					System.currentTimeMillis() - time));
			Db.release();
		}
	}

	@Override
	public <T> void setHandler(Class<T> clazz, JobHandler<T> handler) {
		handlers.put(clazz, handler);
	}

	public JobHandler<?> getHandler(Class<?> clazz) {
		return handlers.get(clazz);
	}

}
